package com.example.rocket.consumer;

import com.example.rocket.core.NbMsg;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.util.Assert;

import java.util.List;
import java.util.Objects;

/**
 * @author TAO
 * @description: RocketMQ-Consumer
 * @date 2021/12/22 4:04 下午
 */

@Slf4j
@Data
public class RocketNbMQConsumer extends AbstractNbMQConsumer {

    private DefaultMQPushConsumer consumer;


    @Override
    public void start() {
        log.info("Rocket===>start");
        if (this.isRunning()) {
            throw new IllegalStateException("容器已经运行！！！ " + this.toString());
        }
        try {
            consumer.start();
        } catch (MQClientException e) {
            throw new IllegalStateException("RocketMQ失败！！！", e);
        }
        this.setRunning(true);

        log.info("RocketMQ运行成功！！！ {}", this.toString());
    }

    @Override
    public void stop() {
        log.info("Rocket===>stop");
        if (this.isRunning()) {
            if (Objects.nonNull(consumer)) {
                consumer.shutdown();
            }
            setRunning(false);
        }
    }

    @Override
    public void destroy(){
        this.setRunning(false);
        if (Objects.nonNull(consumer)) {
            consumer.shutdown();
        }
        log.info("Rocket监听器容器 销毁, {}", this.toString());
    }


    /**
     * 初始化 MQ 消息监听器
     *
     * @throws MQClientException
     */
    protected void initNbMQPushConsumer() throws MQClientException {

        if (nbMqListener == null) {
            throw new IllegalArgumentException("MQ消息消息监听器初始化需要nbMqListener");
        }
        Assert.notNull(consumerGroup, "属性“consumerGroup”是必需的");
        Assert.notNull(nameServer, "属性“nameServer”是必需的");
        Assert.notNull(topic, "属性“topic”是必需的");

        //RPCHook rpcHook = RocketMQUtil.getRPCHookByAkSk(applicationContext.getEnvironment(),
        //        this.rocketMQMessageListener.accessKey(), this.rocketMQMessageListener.secretKey());



        consumer = new DefaultMQPushConsumer();
        consumer.setNamesrvAddr(nameServer);
        consumer.setAccessChannel(accessChannel);
        consumer.setConsumerGroup(consumerGroup);
        consumer.subscribe(topic, selectorExpression);
        consumer.setConsumeThreadMax(consumeThreadMax);
        if (consumeThreadMax < consumer.getConsumeThreadMin()) {
            consumer.setConsumeThreadMin(consumeThreadMax);
        }
        consumer.setConsumeTimeout(consumeTimeout);
        //TODO 消费重试次数
        consumer.setMaxReconsumeTimes(3);

        consumer.setMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                for (MessageExt messageExt : msgs) {
                    log.debug("received msg: {}", messageExt);
                    try {
                        long now = System.currentTimeMillis();
                        handleMessage(new NbMsg(messageExt));
                        long costTime = System.currentTimeMillis() - now;
                        log.debug("consume {} cost: {} ms", messageExt.getMsgId(), costTime);
                    } catch (Exception e) {
                        log.warn("consume message failed. messageExt:{}, error:{}", messageExt, e);
                        //context.setDelayLevelWhenNextConsume(delayLevelWhenNextConsume);
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }

                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });//设置监听器
    }



}
